package com.renrendoc.network.flow.ana;

import com.renrendoc.network.flow.beans.out.PageEvent;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 聚合输出操作
 */
public class PageAnaProcess extends ProcessWindowFunction<PageEvent, PageEvent, String, TimeWindow> {
    @Override
    public void process(String s, ProcessWindowFunction<PageEvent, PageEvent, String, TimeWindow>.Context context, Iterable<PageEvent> elements, Collector<PageEvent> out) throws Exception {
        long start = context.window().getStart();
        long end = context.window().getEnd();
        PageEvent page = elements.iterator().next();
        page.pageIdent = s;
        page.startTime = start;
        page.endTime = end;
        out.collect(page);
    }
}
